package com.chief.sink.redis;


import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.junit.Test;
import redis.clients.jedis.HostAndPort;
import redis.clients.jedis.JedisCluster;

import java.util.*;

public class RedisTest {


    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> streamSource = env.addSource(new RedisSource());

        SingleOutputStreamOperator<String> map = streamSource.map(new MapFunction<String, String>() {
            @Override
            public String map(String value) throws Exception {
                return "map val is " + value;
            }
        });

        map.print();

        map.addSink(new RedisSink());

        env.execute("redis sink1");
    }


    @Test
    public void prepareData() throws InterruptedException {
        JedisCluster jc = getJedisCluster();
        for (int i = 0; i < 1000; i++) {
            jc.set("index_" + i, "value_" + i);
        }
        jc.close();
    }

    public static JedisCluster getJedisCluster() {
        Set<HostAndPort> jedisClusterNodes = new HashSet<>();
        jedisClusterNodes.add(new HostAndPort("192.168.10.52", 6379));
        jedisClusterNodes.add(new HostAndPort("192.168.10.53", 6379));
        jedisClusterNodes.add(new HostAndPort("192.168.10.54", 6379));
        jedisClusterNodes.add(new HostAndPort("192.168.10.54", 6380));
        jedisClusterNodes.add(new HostAndPort("192.168.10.55", 6379));
        jedisClusterNodes.add(new HostAndPort("192.168.10.55", 6380));
        return new JedisCluster(jedisClusterNodes);
    }

    @Test
    public void testGet() throws InterruptedException {
        JedisCluster jc = getJedisCluster();
        for (int i = 0; i < 1000; i++) {
            System.out.println(jc.get("index_" + i));
            Thread.sleep(1000);
        }
        jc.close();
    }
}
